package cn.vetech.sync.elasticsearch.service.impl;

import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import javax.annotation.Resource;

import cn.vetech.sync.elasticsearch.util.DateUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import cn.vetech.sync.elasticsearch.dao.BaseDao;
import cn.vetech.sync.elasticsearch.model.DatabaseTableModel;
import cn.vetech.sync.elasticsearch.model.IndexTypeModel;
import cn.vetech.sync.elasticsearch.model.request.SyncByTableNoPkIntervalRequest;
import cn.vetech.sync.elasticsearch.model.request.SyncByTableRequest;
import cn.vetech.sync.elasticsearch.service.ElasticsearchService;
import cn.vetech.sync.elasticsearch.service.MappingService;
import cn.vetech.sync.elasticsearch.service.SyncService;

/**
 * @author chenqingsong
 * @version 1.0
 * @since 2018-2-7 17:53:07
 */
@Service
public class SyncServiceImpl implements SyncService, InitializingBean, DisposableBean {
    private static final Logger logger = LoggerFactory.getLogger(SyncServiceImpl.class);
    /**
     * 使用线程池控制并发数量
     */
    private ExecutorService cachedThreadPool;

    @Resource
    private BaseDao baseDao;

    @Resource
    private ElasticsearchService elasticsearchService;

    @Resource
    private MappingService mappingService;

    @Override
    public boolean syncByTable(SyncByTableRequest request) {
        // 获取配置数据
        IndexTypeModel indexTypeModel = getIndexTypeModel(request);
        // 获取主键数据
        String primaryKey = getPrimaryKey(request);
        // 检查配置数据是否存在
        checkIndexTypeModel(request, indexTypeModel);
        // 获取任务
        Runnable runnable = getRunnable(request, indexTypeModel, primaryKey);
        // 提交执行任务
        cachedThreadPool.submit(runnable);
        return true;
    }

    /**
     * 根据请求参数生成任务
     *
     * @param request        SyncByTableRequest
     *                       可能是SyncByTableRequest或者SyncByTableNoPkIntervalRequest
     * @param indexTypeModel IndexTypeModel 配置文件数据
     * @param primaryKey     String 主键名称
     * @return Runnable 任务对象
     */
    private Runnable getRunnable(SyncByTableRequest request, IndexTypeModel indexTypeModel, String primaryKey) {
        if (SyncByTableRequest.class == request.getClass()) {
            return getRunnableByPkInterval(request, indexTypeModel, primaryKey);
        } else if (SyncByTableNoPkIntervalRequest.class == request.getClass()) {
            return getRunnableNoPkInterval((SyncByTableNoPkIntervalRequest) request, indexTypeModel, primaryKey);
        } else {
            return null;
        }
    }

    /**
     * 生成带主键的区间任务
     *
     * @param request        SyncByTableRequest 同步表请求数据
     * @param indexTypeModel IndexTypeModel 配置文件
     * @param primaryKey     String 主键
     * @return Runnable 线程
     */
    private Runnable getRunnableByPkInterval(SyncByTableRequest request, IndexTypeModel indexTypeModel,
                                             String primaryKey) {
        long minPK = Optional.ofNullable(request.getFrom())
                .orElse(baseDao.selectMinPK(primaryKey, request.getDatabase(), request.getTable()));
        long maxPK = Optional.ofNullable(request.getTo())
                .orElse(baseDao.selectMaxPK(primaryKey, request.getDatabase(), request.getTable()));
        Runnable runnable = () -> {
            try {
                for (long i = minPK; i < maxPK; i += request.getStepSize()) {
                    List<Map<String, Object>> dataList = baseDao.selectByPKIntervalLockInShareMode(primaryKey, i,
                            (i + request.getStepSize()), request.getDatabase(), request.getTable());
                    batchInsertElasticsearch(request, primaryKey, dataList, indexTypeModel);
                }
            } catch (Exception e) {
                logger.error("批量转换并插入Elasticsearch异常", e);
            }
        };
        return runnable;
    }

    /**
     * 生产未包含主键区间的任务
     *
     * @param request
     * @param indexTypeModel
     * @param primaryKey
     * @return Runnable 线程
     */
    private Runnable getRunnableNoPkInterval(SyncByTableNoPkIntervalRequest request, IndexTypeModel indexTypeModel,
                                             String primaryKey) {
        Runnable runnable = () -> {
            try {
                String minPk = null;//最小主键,即list里面最后一位数的主键
                List<Map<String, Object>> dataList;
                do {
                    //查询所有
                    dataList = baseDao.selectDatasLockInShareMode(primaryKey,
                            ((request.getPageNum() - 1) * request.getPageSize()), (request.getPageNum() * request.getPageSize()), minPk, (request.getDatabase()), request.getTable());
                    //寻找最后一位数据
                    if (dataList != null&&dataList.size()!=0) {
                        minPk = dataList.get(dataList.size() - 1).get(primaryKey).toString();
                    }
                    batchInsertElasticsearch(request, primaryKey, dataList, indexTypeModel);
                } while (null != dataList && dataList.size() >= request.getPageSize());
            } catch (Exception e) {
                logger.error("批量转换并插入Elasticsearch异常", e);
            }
        };
        return runnable;
    }

    /**
     * 检查配置文件
     *
     * @param request        SyncByTableRequest 同步表请求
     * @param indexTypeModel IndexTypeModel 配置数据
     */
    private void checkIndexTypeModel(SyncByTableRequest request, IndexTypeModel indexTypeModel) {
        if (indexTypeModel == null) {
            throw new IllegalArgumentException(String.format("配置文件中缺失database=%s和table=%s所对应的index和type的映射配置",
                    request.getDatabase(), request.getTable()));
        }
    }

    /**
     * 根据请求获取配置数据
     *
     * @param request SyncByTableRequest 同步表请求
     * @return IndexTypeModel 配置数据
     */
    private IndexTypeModel getIndexTypeModel(SyncByTableRequest request) {
        return mappingService.getIndexType(new DatabaseTableModel(request.getDatabase(), request.getTable()));
    }

    /**
     * 获取主键数据
     *
     * @param request SyncByTableRequest 请求数据
     * @return String 表主键名称
     */
    private String getPrimaryKey(SyncByTableRequest request) {
        String primaryKey = Optional
                .ofNullable(
                        mappingService.getTablePrimaryKeyMap().get(request.getDatabase() + "." + request.getTable()))
                .orElse("id");
        return primaryKey;
    }

    /**
     * 批量同步表至es
     *
     * @param request        SyncByTableRequest 请求数据
     * @param primaryKey     String 主键名称
     * @param indexTypeModel IndexTypeModel 配置数据
     */
    @Transactional(readOnly = true, rollbackFor = Exception.class)
    @Override
    public void batchInsertElasticsearch(SyncByTableRequest request, String primaryKey,
                                         List<Map<String, Object>> dataList, IndexTypeModel indexTypeModel) {
        if (dataList == null || dataList.isEmpty()) {
            return;
        }
        handleDateList(primaryKey, indexTypeModel, dataList);
    }

    /**
     * 处理数据至ES
     *
     * @param primaryKey     String 主键名称
     * @param indexTypeModel IndexTypeModel 配置数据
     * @param dataList       List<Map<String, Object>> 数据库数据
     */
    private void handleDateList(String primaryKey, IndexTypeModel indexTypeModel, List<Map<String, Object>> dataList) {
        dataList = convertDateType(dataList);
        Map<String, Map<String, Object>> dataMap = dataList.parallelStream()
                .collect(Collectors.toMap(strObjMap -> String.valueOf(strObjMap.get(primaryKey)), map -> {
                    map.keySet().forEach(key->{
                        Object value=map.get(key);
                        if(null==value){return;}
                        if(LocalDateTime.class==value.getClass()){
                            map.put(key, DateUtil.toDate(value));
                        }
                    });
                    return map;
                }));
        elasticsearchService.batchInsertById(indexTypeModel.getIndex(), indexTypeModel.getType(), dataMap);
    }

    /**
     * 转换Timestamp类型数据
     *
     * @param source List<Map<String, Object>> 表数据
     * @return List<Map   <   String   ,       Object>> 转换后的数据
     */
    private List<Map<String, Object>> convertDateType(List<Map<String, Object>> source) {
        source.parallelStream().forEach(map -> map.forEach((key, value) -> {
            if (value instanceof Timestamp) {
                map.put(key, LocalDateTime.ofInstant(((Timestamp) value).toInstant(), ZoneId.systemDefault()));
            }
        }));
        return source;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        cachedThreadPool = new ThreadPoolExecutor(5, 5, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(),
                (ThreadFactory) Thread::new);
    }

    @Override
    public void destroy() throws Exception {
        if (cachedThreadPool != null) {
            cachedThreadPool.shutdown();
        }
    }

    /**
     * 忽略主键的同步
     */
    @Override
    @Transactional(readOnly = true, rollbackFor = Exception.class)
    public boolean syncTableIgnorePrimaryKey(SyncByTableNoPkIntervalRequest request) {
        return syncByTable(request);
    }

    @Override
    public Long getTableCount(String database, String table) {
        return baseDao.count(database, table);
    }

    @Override
    public List<Map<String, Object>> getTableData(String table_name) {
        return baseDao.getTableData(table_name);
    }

    @Override
    public Boolean clean(SyncByTableNoPkIntervalRequest request) {
        return elasticsearchService.deleteIndex(request.getDatabase(),request.getTable());
    }

}
